package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 对虚拟主机的管理
 */
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    // 操作队列的锁对象
    private final Object queueLocker = new Object();
    //操作绑定的锁对象
    private final Object bindingLocker = new Object();

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public VirtualHost(String name) {
        this.virtualHostName = name;
        //MemoryDataCenter不需要初始化 之后new对象的时候会自动初始化
        //对DiskDataCenter进行初始化
        diskDataCenter.init();

        //将硬盘中的数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败");
        }

    }


    /**
     * 创建交换机
     *
     * @param exchangeName 交换机的名称
     * @param exchangeType 交换机类型
     * @param durable      是否可持久化
     * @param autoDelete   是否能自动删除
     * @param arguments    参数
     * @return
     */
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                   boolean autoDelete, Map<String, Object> arguments) {
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                //查看交换机是否已存在 存在就说明信息
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange != null) {
                    System.out.println("[VirtualHost]交换机已存在" + exchangeName);
                    return true;
                }
                //不存在 开始新建交换机
                exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setDurable(durable);
                exchange.setType(exchangeType);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                //判断durable是否为true
                if (durable) {
                    //将交换机写入硬盘
                    diskDataCenter.insertExchange(exchange);

                }
                //将交换机写入内存
                memoryDataCenter.insertExchange(exchange);

            }
            return true;

        } catch (Exception e) {
            System.out.println("[VirtualHost]创建交换机失败");
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除交换机
     *
     * @param exchangeName 交换机名称
     * @return 返回是否已删除
     * @throws MqException 自定义异常类
     */
    public boolean exchangeDelete(String exchangeName) throws MqException {
        exchangeName = virtualHostName + exchangeName;


        try {
            synchronized (exchangeLocker) {
                //判断交换机是否存在
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost] 交换机不存在 删除失败 ");
                }
                //执行删除操作
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost]交换机删除成功 exchangeName=" + exchangeName);

            }
            return true;

        } catch (Exception e) {
            System.out.println("[VirtualHost]交换机删除失败 exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 创建队列
     *
     * @param queueName  队列名
     * @param durable    是否可持久化
     * @param exclusive  是否独占
     * @param autoDelete 能否自动删除
     * @param arguments  参数
     * @return
     */
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String, Object> arguments) {
        //1.设置队列名字
        queueName = virtualHostName + queueName;
        //2.判断队列是否为空

        try {
            synchronized (queueLocker) {
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue != null) {
                    System.out.println("[VirtualHost]队列不为空 创建失败");
                    return true;
                }
                //3.创建队列对象
                queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //4.写入硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //5.写入内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功 queueName" + queueName);

            }

            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列创建失败 queueName" + queueName);
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 删除队列
     *
     * @param queueName 队列名
     * @return 是否删除
     * @throws IOException
     */


    // 删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                // 1. 根据队列名字, 查询下当前的队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在! 无法删除! queueName=" + queueName);
                }
                // 2. 删除硬盘数据
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3. 删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除队列失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    public boolean queueDelete0(String queueName) throws IOException {
        queueName = virtualHostName + queueName;
        //判断队列是否存在
        try {
            synchronized (queueLocker) {
                MSGQueue toDelete = memoryDataCenter.getQueue(queueName);
                if (toDelete == null) {
                    throw new MqException("[VirtualHost]队列不存在 无法删除");
                }
                //删除队列   看一下是否是持久化的
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost]队列删除成功 queueName" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost]队列删除失败 queueName" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 创建绑定
     *
     * @param exchangeName 交换机名称
     * @param queueName    队列名
     * @param bindingKey   绑定值
     * @return 是否创建绑定成功
     * @throws MqException
     */


    public boolean queueBind0(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 判定当前的绑定是否已经存在了.
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("[VirtualHost] binding 已经存在! queueName=" + queueName
                                + ", exchangeName=" + exchangeName);
                    }
                    // 2. 验证 bindingKey 是否合法.
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("[VirtualHost] bindingKey 非法! bindingKey=" + bindingKey);
                    }
                    // 3. 创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取一下对应的交换机和队列. 如果交换机或者队列不存在, 这样的绑定也是无法创建的.
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[VirtualHost] 队列不存在! queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName=" + exchangeName);
                    }
                    // 5. 先写硬盘
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 写入内存
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }



    public boolean queueBind(String exchangeName, String queueName, String bindingKey) throws MqException {
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (bindingKey) {
                //绑定是否存在了
                Binding exsistBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                if (exsistBinding != null) {
                    throw new MqException("[VirtualHost] 创建绑定失败！绑定已存在");
                }
                //key是否有效
                if (!router.checkBindingKey(bindingKey)) {
                    throw new MqException("[VirtualHost] bingdingKey非法 bindingKey = " + bindingKey);
                }
                //创建binding

                Binding binding = new Binding();
                binding.setExchangeName(exchangeName);
                binding.setQueueName(queueName);
                binding.setBindingKey(bindingKey);
                //exchange和queue都在吗 缺一不可绑定
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] queueName不存在 queueName= " + queueName);
                }
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if (exchange == null) {
                    throw new MqException("[VirtualHost] exchangeName无效 exchangeName= " + exchangeName);
                }
                //写入硬盘 内存
                if (queue.isDurable() && exchange.isDurable()) {
                    diskDataCenter.insertBinding(binding);
                }
                memoryDataCenter.insertBinding(binding);


            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName=" + exchangeName
                    + ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建绑定失败 exchangeName=" + exchangeName
                    + "queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除绑定
     *
     * @param queueName    队列名
     * @param exchangeName 交换机名
     * @return
     */
    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    // 1. 获取 binding
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    // 2. 无论绑定是否持久化了, 都尝试从硬盘删一下. 就算不存在, 这个删除也无副作用.
                    diskDataCenter.deleteBinding(binding);
                    // 3. 删除内存的数据
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功!");
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 发送消息到指定队列
     *
     * @param exchangeName    交换机名称
     * @param routingKey      路由关键字
     * @param basicProperties 消息相关属性
     * @param body            消息体
     * @return 是否发送成功
     * @throws MqException
     */
    public boolean basicPublish(String exchangeName, String routingKey,
                                BasicProperties basicProperties, byte[] body) throws MqException {
        try {
            //设置交换机的名称
            exchangeName = virtualHostName + exchangeName;
            //查看交换机是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey非法 rountingKey=" + routingKey);
            }

            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
            }
            //判断交换机类型
            //直接交换机
            if (exchange.getType() == ExchangeType.DIRECT) {

                String queueName = virtualHostName + routingKey;
                //构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                //根据对列名查找存储该消息的队列对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在 queueName=" + queueName);

                }
                //查找到则进行发送
                sendMessage(queue, message);
            } else {
                //按照fanout和topic的方式转发
                //先获取这个交换机所有的绑定，再遍历绑定的所有对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    // 1) 获取到绑定对象, 判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName=" + binding.getQueueName());
                    }
                    //构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    //判断是否可以发送
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    sendMessage(queue, message);
                }
            }
            return true;

        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 发送消息
     *
     * @param queue   要发送到的队列
     * @param message 要发送的消息
     * @throws IOException
     * @throws MqException
     */
    public void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getDeliverMode();
        //deliverMode为1 不持久化 为2初九话
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        memoryDataCenter.sendMessage(message, queue);
        //  通知消费者可以消费消息了.
        consumerManager.notifyConsume(queue.getName());

    }


    /**
     * 订阅者队列
     *
     * @param consumerTag 消费者标识
     * @param queueName   所订阅的队列名
     * @param autoAck     是否自动确认应答
     * @param consumer    消费者
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        // 构造一个 ConsumerEnv 对象, 把这个对应的队列找到, 再把这个 Consumer 对象添加到该队列中.
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功! queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;

        }

    }

    /**
     * 手动确认消息
     *
     * @param queueName 队列名
     * @param messageId 消息Id
     * @return 是否手动确认消息成功
     * @throws MqException
     * @throws IOException
     * @throws ClassNotFoundException
     */
    public boolean basicAck(String queueName, String messageId) throws MqException, IOException, ClassNotFoundException {
        queueName = virtualHostName + queueName;
        try {        //获取消息和队列对象
            Message message = memoryDataCenter.getMessage(messageId);
            MSGQueue queue = memoryDataCenter.getQueue(queueName);

            //判空
            if (message == null) {
                throw new MqException("[VirtualHost] 消息不存在 messageId=" + messageId);
            }
            if (queue == null) {
                throw new MqException("[VirtualHost] 队列不存在 queueName=" + queueName);
            }
            //删除 硬盘
            if (message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            //删除 内存中的message
            memoryDataCenter.removeMessage(messageId);
            //删除 待确认队列中的message
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck成功 消息已成功确认 queueName=" + queueName + "messageId=" + messageId);
            return true;

        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck成功 消息确认失败 queueName=" + queueName + "messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }


}
